package com.aotain.jupiter.cc;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import com.aotain.jupiter.common.URLStatTuple;

public class URLStatReduce implements ReduceFunction<Tuple2<String, URLStatTuple>>{

	/**
	 * 
	 */
	private static final long serialVersionUID = 7029126205053500825L;

	@Override
	public Tuple2<String, URLStatTuple> reduce(Tuple2<String, URLStatTuple> arg0,
			Tuple2<String, URLStatTuple> arg1) throws Exception {
		// TODO Auto-generated method stub
		
		URLStatTuple info = new URLStatTuple();
		
		if(arg0.f1 != null && arg1.f1 != null) {
			info.setHTTPNum(arg0.f1.getHTTPNum() + arg1.f1.getHTTPNum());
			info.setUpStreamOctets(arg0.f1.getUpStreamOctets() + arg1.f1.getUpStreamOctets());
			info.setUpStreamPacket(arg0.f1.getUpStreamPacket() + arg1.f1.getUpStreamPacket());
		} else {
			System.out.println("reduce arg is null");
		}
		return new Tuple2<String, URLStatTuple> (arg0.f0,info);
	}

}
